from xtquant import xtdata
from datetime import datetime
import time
from typing import Dict, List
import pandas as pd

from 新版2.系统组件.基础配置 import settings
from 新版2.指标计算.MACD指标 import MACDSignalCalculator
from 新版2.指标计算.指标配置 import MACDConfig
from 新版2.指标计算.RSI指标 import THSRSICalculator
from 新版2.指标计算.日内九转 import THSNineTurn
from 新版2.指标计算.振幅指标 import UpDown
from 新版2.指标计算.波动率 import ATRCalculator
from collections import deque
from 新版2.指标计算.成交量 import solution

class MarketMonitor:
    def __init__(self):
        self._running = False
        self.stock_codes = [s["all_code"] for s in settings.STOCK_CODE]
        self.historical_ticks: Dict[str, object] = {}
        # 指标实例
        cfg = MACDConfig.get_optimized_params()
        self.macd_calculator = MACDSignalCalculator(xtdata, use_warmup=cfg['use_warmup'])
        self.macd_fast = MACDConfig.FAST_EMA
        self.macd_slow = MACDConfig.SLOW_EMA
        self.macd_signal = MACDConfig.SIGNAL_EMA
        self.tick_counters = {code: 0 for code in self.stock_codes}
        self.macd_cache = {}
        self.last_macd_check = {code: 0 for code in self.stock_codes}
        self.macd_check_interval = MACDConfig.TICK_CHECK_INTERVAL
        # 严格同花顺口径：仅按收盘价更新的 RSI10
        self.rsi_calculator = THSRSICalculator(period=10)
        # 分钟聚合：记录当前分钟与该分钟的最新价作为收盘
        self._minute_bucket: Dict[str, str] = {}
        self._minute_close: Dict[str, float] = {}
        self.updown = UpDown()
        self.nine_turn = THSNineTurn(compare_offset=4, required_bars=10, verbose=True, restrict_session=True)
        self.atr_calculator = ATRCalculator(xtdata, lookback=getattr(settings, 'ATR_LOOKBACK', 14), method='ema', cache_ttl_seconds=10, verbose=False)
        # 价格历史用于EMA
        self.price_history = {code: deque(maxlen=100) for code in self.stock_codes}
        # 指标状态（可用于外部读取）
        self.indicator_state: Dict[str, Dict[str, object]] = {code: {} for code in self.stock_codes}
        # 成交量放大检测（VMA）
        self.solution = solution()
        # 每只股票上一笔“累计成交量”
        self._last_cum_volume: Dict[str, float] = {}

    def _is_session_minute(self, minute_str: str) -> bool:
        try:
            if not minute_str:
                return False
            dt = datetime.strptime(minute_str, '%Y-%m-%d %H:%M')
            t = dt.time()
            m_start = datetime.strptime("09:30", "%H:%M").time()
            m_end = datetime.strptime("11:30", "%H:%M").time()
            a_start = datetime.strptime("13:00", "%H:%M").time()
            a_end = datetime.strptime("15:00", "%H:%M").time()
            return (m_start <= t <= m_end) or (a_start <= t <= a_end)
        except Exception:
            return False

    ##订阅tick回调
    def quote_callback(self, data):
        for stock_code, stock_data in data.items():
            # 合并历史
            self.add_to_history(stock_code, stock_data)
            # 解析最新一笔数据
            last_price, timestamp = self._extract_last_price_time(stock_data)
            # 成交量：将可能是“累计量”的字段换算为“每tick增量”，并进行分钟聚合
            try:
                vol = self._extract_volume(stock_data)
                vol_f = float(vol) if vol is not None else 0.0
                last_cum = self._last_cum_volume.get(stock_code)
                if last_cum is not None and vol_f >= last_cum:
                    per_tick_vol = vol_f - last_cum
                else:
                    per_tick_vol = vol_f
                self._last_cum_volume[stock_code] = vol_f
                self.solution.on_tick(stock_code, timestamp, per_tick_vol)
            except Exception:
                pass
            if last_price is None or timestamp is None:
                continue
            # 维护tick计数
            self.tick_counters.setdefault(stock_code, 0)
            self.tick_counters[stock_code] += 1
            # 价格历史
            try:
                self.price_history.setdefault(stock_code, deque(maxlen=100)).append(float(last_price))
            except Exception:
                pass
            # 分钟收盘聚合：同花顺 RSI 仅在分钟/日线收盘后更新
            try:
                minute_key = self._to_minute_bucket(timestamp)
                if minute_key is not None:
                    prev_key = self._minute_bucket.get(stock_code)
                    if prev_key is None:
                        self._minute_bucket[stock_code] = minute_key
                        self._minute_close[stock_code] = float(last_price)
                    elif prev_key == minute_key:
                        # 同一分钟内，更新该分钟的“收盘价”为最新 tick 价
                        self._minute_close[stock_code] = float(last_price)
                    else:
                        # 分钟切换：用上一分钟的最后价作为“收盘价”更新 RSI（仅交易时段内的上一分钟）
                        prev_close = self._minute_close.get(stock_code)
                        if prev_close is not None and self._is_session_minute(prev_key):
                            try:
                                self.rsi_calculator.update_with_close(stock_code, prev_close)
                            except Exception:
                                pass
                        self._minute_bucket[stock_code] = minute_key
                        self._minute_close[stock_code] = float(last_price)
            except Exception:
                pass

            # 计算指标
            self.get_indicator(stock_code, last_price, timestamp, stock_data)

    ##根据每根tick计算指标
    def get_indicator(self, stock_code, last_price, timestamp, stock_data):
        # RSI（严格同花顺口径）：优先使用“当前分钟预览值”，否则取最近已确认分钟值
        try:
            rsi_value = self.rsi_calculator.get_preview_rsi(stock_code, float(last_price))
        except Exception:
            rsi_value = None
        if rsi_value is None:
            rsi_value = self.rsi_calculator.get_last_rsi(stock_code)
        # 九转
        nine_sig = self.nine_turn.on_tick(stock_code, timestamp, last_price)
        # 振幅（横盘）
        try:
            updown_id = self.updown.updown_watch_tick(stock_code, stock_data)
        except Exception:
            updown_id = 0
        # MACD（按间隔触发）
        macd_sig = None
        try:
            if self.tick_counters.get(stock_code, 0) % self.macd_check_interval == 0:
                macd_signals = self.macd_calculator.detect_macd_signals(
                    stock_code,
                    fast=self.macd_fast,
                    slow=self.macd_slow,
                    signal=self.macd_signal,
                    use_warmup=MACDConfig.USE_WARMUP
                )
                macd_sig = self._process_macd_signals_quick(macd_signals)
        except Exception:
            macd_sig = None

        # 买卖判定
        action, reasons, filters = self.get_buy_or_sell(stock_code, last_price, macd_sig, rsi_value, nine_sig)
        # 存入状态
        self.indicator_state[stock_code] = {
            'rsi': rsi_value,
            'nine_turn': nine_sig,
            'updown_id': updown_id,
            'macd': macd_sig,
            'last_price': last_price,
            'timestamp': timestamp,
            'action': action,
            'reasons': reasons,
            'vma_big': self._safe_vma_flag(stock_code),
            'filters': filters,
        }

    ##根据指标计算买入or卖出
    def get_buy_or_sell(self, stock_code: str, last_price: float, macd_sig: str, rsi_value: float, nine_sig: str):
        need_buy = False
        need_sell = False
        reasons = []
        filters: Dict[str, object] = {}

        # MACD 金叉买入判定（需同时满足3个条件：ATR过滤、RSI<60、VMA放大）
        if macd_sig == '金叉':
            print(f'{stock_code}金叉')
            can_buy = 0
            # ATR过滤（无法获取ATR时默认通过）
            try:
                atr = self.atr_calculator.calculate_atr(stock_code, last_price)
                if atr is not None and last_price:
                    atr_ratio = atr / float(last_price)
                    threshold = max(0.0006, atr_ratio * 0.7)
                    filters['atr'] = float(atr)
                    filters['atr_ratio'] = float(atr_ratio)
                    filters['atr_threshold'] = float(threshold)
                    if atr_ratio > threshold:
                        can_buy += 1
                        filters['atr_ok'] = True
                        print(f"ATR过滤通过 ratio={atr_ratio:.4%} > th={threshold:.4%}")
                    else:
                        filters['atr_ok'] = False
                        print(f"ATR过滤不通过 ratio={atr_ratio:.4%} <= th={threshold:.4%}")
                else:
                    can_buy += 1
                    filters['atr'] = None
                    filters['atr_ratio'] = None
                    filters['atr_threshold'] = None
                    filters['atr_ok'] = True
                    print("ATR缺失，默认通过")
            except Exception:
                can_buy += 1
                filters['atr'] = None
                filters['atr_ratio'] = None
                filters['atr_threshold'] = None
                filters['atr_ok'] = True
                print("ATR异常，默认通过")

            # RSI<60
            try:
                rsi_latest = rsi_value if rsi_value is not None else self.rsi_calculator.get_preview_rsi(stock_code, float(last_price))
                if rsi_latest is None:
                    rsi_latest = self.rsi_calculator.get_last_rsi(stock_code)
                filters['rsi'] = float(rsi_latest) if rsi_latest is not None else None
                if rsi_latest is not None and rsi_latest > 0:
                    if rsi_latest < 60:
                        can_buy += 1
                        filters['rsi_ok'] = True
                        print(f"RSI通过 {rsi_latest:.2f}<60")
                    else:
                        filters['rsi_ok'] = False
                        print(f"RSI不通过 {rsi_latest:.2f}>=60")
                else:
                    filters['rsi_ok'] = False
                    reasons.append("RSI初始化中")
            except Exception:
                filters['rsi_ok'] = False
                print("RSI计算异常")

            # 成交量VMA放大（输出调试信息）。为与MACD金叉“上一分钟触发”对齐，优先采用上一分钟是否放量
            try:
                vma_ok = False
                if hasattr(self.solution, 'was_prev_minute_big') and self.solution.was_prev_minute_big(stock_code):
                    vma_ok = True
                elif self.solution.make_big(stock_code):
                    vma_ok = True
                if vma_ok:
                    can_buy += 1
                    print(f"VMA {stock_code} 通过：上一/当前分钟放量")
                    filters['vma_big'] = True
                    filters['vma_ok'] = True
                else:
                    filters['vma_big'] = False
                    filters['vma_ok'] = False
                    print(f"VMA {stock_code} 不通过：成交量未放大")
            except Exception:
                filters['vma_big'] = None
                filters['vma_ok'] = False
                print("VMA计算异常")

            # EMA5
            # ema5 = self._calculate_ema(list(self.price_history.get(stock_code, [])), 5)
            # if ema5 is not None and last_price is not None:
            #     try:
            #         if float(last_price) > float(ema5):
            #             can_buy += 1
            #             reasons.append(f"EMA5通过 price>{ema5:.6f}")
            #         else:
            #             reasons.append(f"EMA5不通过 price<={ema5:.6f}")
            #     except Exception:
            #         reasons.append("EMA比较异常")
            # else:
            #     reasons.append("EMA5数据不足")
            #
            # print(reasons)

            if can_buy == 3:
                need_buy = True

        # MACD 死叉卖出
        if macd_sig == '死叉':
            need_sell = True

        # 九转信号直接映射
        if nine_sig == '买入':
            need_buy = True
            reasons.append('九转买')
        if nine_sig == '卖出':
            need_sell = True
            reasons.append('九转卖')

        filters['can_buy_count'] = int(can_buy) if 'can_buy' in locals() else 0
        action = 'buy' if need_buy and not need_sell else ('sell' if need_sell and not need_buy else None)
        return action, reasons, filters



    ##订阅数据合并到历史数据
    def add_to_history(self,stock_code,stock_data):
        try:
            # 将收到的数据转为DataFrame
            if stock_data is None:
                return
            if isinstance(stock_data, list):
                if not stock_data:
                    return
                df_new = pd.DataFrame(stock_data)
            elif hasattr(stock_data, 'empty'):
                df_new = stock_data.copy()
            else:
                return

            # 规范化列，并生成可用于去重/排序的 time 列
            df_new = self._normalize_ticks_df(df_new)
            if df_new.empty:
                return

            # 合并到历史缓存
            df_old = self.historical_ticks.get(stock_code)
            if df_old is None or (hasattr(df_old, 'empty') and df_old.empty):
                self.historical_ticks[stock_code] = df_new
            else:
                df_all = pd.concat([df_old, df_new], ignore_index=True, sort=False)
                if 'time' in df_all.columns:
                    df_all.drop_duplicates(subset=['time'], keep='last', inplace=True)
                    df_all.sort_values('time', inplace=True)
                else:
                    df_all.drop_duplicates(inplace=True)
                df_all.reset_index(drop=True, inplace=True)
                self.historical_ticks[stock_code] = df_all

        except Exception as e:
            print(f"[ERR] 回调合并 {stock_code} 失败: {e}")
    
    def _extract_last_price_time(self, stock_data):
        try:
            if stock_data is None:
                return None, None
            if isinstance(stock_data, list):
                if not stock_data:
                    return None, None
                d0 = stock_data[0]
            elif hasattr(stock_data, 'empty'):
                if stock_data.empty:
                    return None, None
                d0 = stock_data.iloc[-1].to_dict()
            elif isinstance(stock_data, dict):
                d0 = stock_data
            else:
                return None, None
            last_price = d0.get('lastPrice') or d0.get('price') or d0.get('close') or d0.get('last') or d0.get('current')
            ts = d0.get('time') or d0.get('timestamp')
            return last_price, ts
        except Exception:
            return None, None

    def _process_macd_signals_quick(self, macd_signals: Dict[str, object]):
        try:
            if macd_signals.get('golden_cross'):
                if len(macd_signals['golden_cross']) > 0:
                    return '金叉'
            if macd_signals.get('death_cross'):
                if len(macd_signals['death_cross']) > 0:
                    return '死叉'
        except Exception:
            pass
        return None
    ##规范化列，并生成可用于去重/排序的 time 列
    def _normalize_ticks_df(self, df: pd.DataFrame) -> pd.DataFrame:
        if df is None or df.empty:
            return pd.DataFrame()
        df = df.copy()
        # 识别时间列
        time_col = next((c for c in df.columns if 'time' in str(c).lower()), None)
        if time_col is not None:
            try:
                # 常见情况：毫秒/秒时间戳或形如 20250101101530 的数值/字符串
                if pd.api.types.is_integer_dtype(df[time_col]) or pd.api.types.is_float_dtype(df[time_col]):
                    # 假设毫秒
                    dt = pd.to_datetime(df[time_col], unit='ms', errors='coerce')
                    # 如果多数为NaT，退回按秒
                    if dt.isna().mean() > 0.5:
                        dt = pd.to_datetime(df[time_col], unit='s', errors='coerce')
                    df['time'] = dt.dt.tz_localize('UTC', nonexistent='shift_forward', ambiguous='NaT').dt.tz_convert('Asia/Shanghai').dt.tz_localize(None)
                else:
                    # 先尝试数值格式串
                    parsed = pd.to_datetime(df[time_col].astype(str), format='%Y%m%d%H%M%S', errors='coerce')
                    if parsed.isna().mean() > 0.5:
                        parsed = pd.to_datetime(df[time_col], errors='coerce')
                    df['time'] = parsed
            except Exception:
                df['time'] = pd.NaT

        # 保留有效时间
        if 'time' in df.columns:
            df = df[df['time'].notna()].copy()
        return df

    def _to_minute_bucket(self, ts) -> str:
        try:
            if ts is None:
                return None
            # 数值：优先按毫秒，其次按秒
            if isinstance(ts, (int, float)):
                t = float(ts)
                if t > 1e12:  # 毫秒
                    dt = datetime.fromtimestamp(t / 1000)
                else:
                    dt = datetime.fromtimestamp(t)
                return dt.strftime('%Y-%m-%d %H:%M')
            # 字符串：尝试常见格式
            s = str(ts)
            if s.isdigit() and len(s) == 14:
                # 形如 20250101101530
                dt = datetime.strptime(s, '%Y%m%d%H%M%S')
                return dt.strftime('%Y-%m-%d %H:%M')
            try:
                dt = pd.to_datetime(s, errors='coerce')
                if pd.isna(dt):
                    return None
                # 转为本地 naive
                if getattr(dt, 'tzinfo', None) is not None:
                    dt = dt.tz_convert('Asia/Shanghai').tz_localize(None)
                return dt.strftime('%Y-%m-%d %H:%M')
            except Exception:
                return None
        except Exception:
            return None

    def _extract_volume(self, stock_data):
        try:
            if stock_data is None:
                return None
            if isinstance(stock_data, list):
                if not stock_data:
                    return None
                d0 = stock_data[-1] if isinstance(stock_data[-1], dict) else stock_data[0]
            elif hasattr(stock_data, 'empty'):
                if stock_data.empty:
                    return None
                d0 = stock_data.iloc[-1].to_dict()
            elif isinstance(stock_data, dict):
                d0 = stock_data
            else:
                return None
            return d0.get('volume') or d0.get('vol') or d0.get('成交量')
        except Exception:
            return None

    def _safe_vma_flag(self, stock_code: str):
        try:
            return bool(self.solution.make_big(stock_code))
        except Exception:
            return None

    def _calculate_ema(self, prices, period: int):
        try:
            if not prices or len(prices) < period:
                return None
            k = 2 / (period + 1)
            ema = sum(prices[:period]) / period
            for price in prices[period:]:
                ema = price * k + ema * (1 - k)
            return float(ema)
        except Exception:
            return None
    ## 订阅前补全：下载并缓存当日全量tick到本地，供各指标读取
    def prefetch_today_ticks(self, codes: List[str] = None):
        codes = codes or self.stock_codes
        if not codes:
            print("未配置订阅代码，跳过tick预取")
            return

        today = datetime.now().strftime("%Y%m%d")
        print(f"开始预取 {today} 当日tick，共 {len(codes)} 只标的……")

        for code in codes:
            try:
                # 下载当日tick到本地
                xtdata.download_history_data(stock_code=code, period="tick", start_time=today, end_time=today)
                # 略等磁盘落地
                time.sleep(0.5)
                # 读取本地tick
                data = xtdata.get_local_data(stock_list=[code], period='tick', start_time=today, end_time=today)
                if data and code in data and data[code] is not None and not data[code].empty:
                    # 统一通过 add_to_history 进行规范化与合并
                    self.add_to_history(code, data[code])
                    size = len(self.historical_ticks.get(code, []))
                    print(f"[OK] {code} 预取tick {size} 条")
                    # 基于已合并的当日tick进行成交量VMA预热（仅纳入已完成分钟）
                    try:
                        dfh = self.historical_ticks.get(code)
                        if dfh is not None and hasattr(dfh, 'empty') and not dfh.empty and 'time' in dfh.columns:
                            vol_col = next((c for c in ['volume', 'vol', '成交量'] if c in dfh.columns), None)
                            if vol_col is not None:
                                dfv = dfh[['time', vol_col]].copy()
                                minute_bucket = pd.to_datetime(dfv['time']).dt.floor('min')
                                v = pd.to_numeric(dfv[vol_col], errors='coerce').fillna(0.0)
                                diffs = v.diff().fillna(v)
                                # 若大部分为非负，视为累计量
                                if (v.diff().fillna(0) >= 0).mean() > 0.8:
                                    per_tick = diffs.clip(lower=0)
                                else:
                                    per_tick = v.clip(lower=0)
                                minute_vol = per_tick.groupby(minute_bucket).sum()
                                now_floor = pd.to_datetime(datetime.now().strftime('%Y-%m-%d %H:%M'))
                                minute_vol = minute_vol[minute_vol.index < now_floor]
                                if len(minute_vol) > 0:
                                    self.solution.warmup_with_volumes(code, minute_vol.tolist())
                    except Exception:
                        pass
                else:
                    self.historical_ticks[code] = None
                    print(f"[WARN] {code} 当日本地tick为空")
            except Exception as e:
                self.historical_ticks[code] = None
                print(f"[ERR] {code} 预取失败: {e}")
    ##启动
    def start(self):
        print(f"开启实时行情")
        self._running = True
        # 订阅前先做当日tick补全
        self.prefetch_today_ticks()

        # 订阅实时行情
        for code in self.stock_codes:
            xtdata.subscribe_quote(
                code,
                period="tick",
                count=-1,
                callback=self.quote_callback
            )
            print(f"已订阅 {code} 的实时行情")
        print(f"开始监听 {len(self.stock_codes)} 只股票的实时行情...")

        try:
            xtdata.run()  # 阻塞
        except KeyboardInterrupt:
            print("行情监控已手动停止")
        except Exception as e:
            print(f"行情监控异常: {e}")
    ##停止
    def stop(self):
        self._running = False
        print("行情监控已停止")


if __name__ == "__main__":
    # 示例运行：
    # 1) 从配置加载标的
    monitor = MarketMonitor()

    # 2) 启动订阅（阻塞运行，可按需注释掉）
    monitor.start()